package com.zyf.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * @author Malegod_xiaofei
 * @create 2024-08-20-21:20
 */
public class KafkaConsumerAutoOffsetTest {
    public static void main(String[] args) {

        // TODO 创建配置对象
        Map<String, Object> consumerConfig = new HashMap();
        consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "zyf");
        consumerConfig.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, false);
        // consumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // TODO 创建消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(consumerConfig);

        // TODO 订阅主题
        consumer.subscribe(Collections.singletonList("test"));

        // TODO 获取集群信息
        boolean flag = true;
        while (flag) {
            consumer.poll(Duration.ofMillis(100));
            Set<TopicPartition> assignment = consumer.assignment();
            if (assignment != null && !assignment.isEmpty()) {
                for (TopicPartition topicPartition : assignment) {
                    if ("test".equals(topicPartition.topic())) {
                        consumer.seek(topicPartition, 2);
                        flag = false;
                    }
                }
            }
        }

        // TODO 从 kafka 的主题中获取数据
        //      消费者从 kafka 中拉取数据
        while (true) {
            final ConsumerRecords<String, String> datas = consumer.poll(Duration.ofMillis(100));

            for (ConsumerRecord<String, String> data : datas) {
                System.out.println(data);
            }

            // TODO 手动保存偏移量
            consumer.commitAsync();// 异步提交
            consumer.commitSync();// 同步提交
        }

        // 关闭消费者对象
        // consumer.close();
    }
}
